/**
 * Copyright [2019] [LiBo/Alex of copyright liboware@gmail.com ]
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hyts.slogger.reactive.base;

import com.hyts.slogger.reactive.config.ReactiveConfiguration;
import com.hyts.threads.core.ThreadPoolManager;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * @project-name:coffice-toolbox
 * @package-name:com.hyts.slogger.reactive.publish
 * @author:LiBo/Alex
 * @create-date:2019-08-06 14:36
 * @copyright:libo-alex4java
 * @email:liboware@gmail.com
 * @description:
 */
public abstract class ReactiveDataPublish<T extends ReactiveLoggerEvent> {

    /**
     * 执行操作处理器服务功能
     */
    private Executor executor;

    /**
     * distruptor对象服务
     */
    private Disruptor<T> disruptor;

    /**
     * ringbuffer对象服务
     */
    private RingBuffer<T> ringBuffer;

    /**
     *
     * @param reactiveDataHandler
     * @param reactiveEventFactory
     */
    public ReactiveDataPublish(ReactiveDataHandler<T> reactiveDataHandler,ReactiveEventFactory<T> reactiveEventFactory) {
        Objects.requireNonNull(reactiveDataHandler);
        disruptor = new Disruptor<T>(reactiveEventFactory, ReactiveConfiguration.DEFAULT_RING_BUFFER_SIZE,
                ThreadPoolManager.getInstance(false).createThreadPool(false));
        disruptor.handleEventsWith(reactiveDataHandler);
        ringBuffer = disruptor.getRingBuffer();
        disruptor.start();
    }

    /**
     * publish-发布信息数据对象
      * @param param
     */
    public ReactiveDataPublish publish(Object param){
        Objects.requireNonNull(param);
        long sequence = ringBuffer.next();
        // Grab the next sequence
        try {
             T event = ringBuffer.get(sequence);
            /**
             * Get the entry in the Disruptor,for the sequence
             * Fill with data
             */
            event.setLogger(param);
        } finally {
            ringBuffer.publish(sequence);
        }
        return this;
    }

}
